package com.atguigu.day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

//通过自定义状态机实现三次错误登录提示
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .fromElements(
                        LoginEvent.of("userId-1", "1.1.1.1", "fail", 1000L),
                        LoginEvent.of("userId-1", "1.1.1.2", "fail", 2000L),
//                        LoginEvent.of("userId-2", "1.1.1.3", "fail", 3000L),
//                        LoginEvent.of("userId-2", "1.1.1.4", "fail", 4000L),
                        LoginEvent.of("userId-1", "1.1.1.5", "fail", 5000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<LoginEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                            @Override
                            public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
                )
                .keyBy(x->x.userId)
                .process(new KeyedProcessFunction<String, LoginEvent, String>() {
                    //声明状态集
                    private final HashMap<Tuple2<String,String>, String> stateMachine = new HashMap<>();
                    private ValueState<String> currentState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        currentState= getRuntimeContext().getState(new ValueStateDescriptor<String>("current-state", Types.STRING));
                        stateMachine.put(Tuple2.of("INITIAL","fail"),"S1");
                        stateMachine.put(Tuple2.of("S1","fail"),"S2");
                        stateMachine.put(Tuple2.of("S2","fail"),"FAIL");
                        stateMachine.put(Tuple2.of("INITIAL","success"),"SUCCESS");
                        stateMachine.put(Tuple2.of("S1","success"),"SUCCESS");
                        stateMachine.put(Tuple2.of("S2","success"),"SUCCESS");
                    }

                    @Override
                    public void processElement(LoginEvent value, Context ctx, Collector<String> out) throws Exception {
                        System.out.println(currentState.value());
                        if (currentState.value() == null) {
                            currentState.update("INITIAL");
                        }
                        String stage = stateMachine.get(Tuple2.of(currentState.value(), value.type));
                        currentState.update(stage);
                        switch (currentState.value()) {
                            case "SUCCESS":
                                currentState.update("INITIAL");
                                break;
                            case "FAIL":
                                currentState.update("INITIAL");
                                out.collect("账号连续3次错误登录" + value.userId);
                                break;
                        }
                    }
                })
                .print();

        env.execute();
    }

    public static class LoginEvent {
        public String userId;
        public String ipAddress;
        public String type;
        public Long timestamp;

        @Override
        public String toString() {
            return "LoginEvent{" +
                    "userId='" + userId + '\'' +
                    ", ipAddress='" + ipAddress + '\'' +
                    ", type='" + type + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }

        public LoginEvent() {
        }

        public static LoginEvent of(String userId, String ipAddress, String type, Long timestamp) {
            return new LoginEvent(userId, ipAddress, type, timestamp);
        }

        private LoginEvent(String userId, String ipAddress, String type, Long timestamp) {
            this.userId = userId;
            this.ipAddress = ipAddress;
            this.type = type;
            this.timestamp = timestamp;
        }
    }
}
